use crate::physical_plan::window::WindowElement;
use std::collections::HashMap;
use std::marker::PhantomData;
use sort::exchange_sorts::quicksort;
use std::cmp::Ordering;
use crate::physical_plan::window::SortByValue;
use std::vec::IntoIter;

#[derive(Clone)]
pub struct WindowGroup<K,V,T>{
    pub element_list: Vec<T>,
    phantom_k: PhantomData<K>,
    phantom_v: PhantomData<V>,
}

impl <K,V,T> WindowGroup<K,V,T>{
    pub fn new()->WindowGroup<K,V,T>{
        WindowGroup{
            element_list: Vec::new(),
            phantom_k: PhantomData,
            phantom_v: PhantomData,
        }
    }
    pub fn push(&mut self, element: T){
        self.element_list.push(element);
    }
}

impl <K,V,T> IntoIterator for WindowGroup<K,V,T>{
    type Item = T;
    type IntoIter = IntoIter<Self::Item>;

    fn into_iter(self) -> Self::IntoIter{
        self.element_list.into_iter()
    }
}

pub enum TimeOutWindowGroup<K,V,T>{
    NoPart(WindowGroup<K,V,T>),
    Part(HashMap<usize,WindowGroup<K,V,T>>),
}

pub type WindowResult<T> = Result<T,()>;

impl <K,V:PartialOrd+PartialEq+Clone,T: WindowElement<K,V>> WindowGroup<K,V,T>{
    pub fn count(&self)->usize{
        self.element_list.len()
    }
    pub fn take_nth(&self, index: usize)->Option<T::MapValue>{
        let nth_idx = self.element_list.get(index)?;
        Some(nth_idx.index_to_value())
    }
    pub fn first(&self)->Option<T::MapValue>{
        self.take_nth(1)
    }
    pub fn last(&self)->Option<T::MapValue>{
        let last_idx = self.element_list.last()?;
        Some(last_idx.index_to_value())
    }
    /*
        根据一组元素对数据进行排序
    */
    pub fn sort_by(&mut self, key: &[K])->WindowResult<Vec<T::MapValue>>{
        let mut value_list = Vec::new();

        for v in self.element_list.iter(){
            let value = v.index_to_value();

            let mut sort_value_list = Vec::new();
            for k in key.iter(){
                let sort_value = value.get_key_value_always(k);

                sort_value_list.push(sort_value);
            }
            value_list.push(Sort::new(sort_value_list, value));
        }

        quicksort(&mut value_list);

        let mut bak = Vec::new();
        for s in value_list.into_iter(){
            bak.push(s.value);
        }

        return WindowResult::Ok(bak);
    }
}

#[derive(Clone)]
struct Sort<V,MapValue>{
    sort_value: Vec<V>,
    value: MapValue,
}

impl <V:PartialEq,MapValue> PartialEq for Sort<V,MapValue>{
    fn eq(&self, other: &Self) -> bool{
        self.sort_value.eq(&other.sort_value)
    }
}

impl <V:PartialEq+PartialOrd,MapValue> PartialOrd for Sort<V,MapValue>{
    fn partial_cmp(&self, other: &Self) -> Option<Ordering>{
        self.sort_value.partial_cmp(&other.sort_value)
    }
}

impl <V,MapValue> Sort<V,MapValue>{
    fn new(sort_value: Vec<V>, value: MapValue)->Sort<V,MapValue>{
        Sort{
            sort_value: sort_value,
            value: value,
        }
    }
}